package com.dongdongshop.mq;

import com.alibaba.fastjson.JSONArray;
import com.dongdongshop.pojo.TbUserScore;
import com.dongdongshop.service.UserScoreService;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;


@Component
@RocketMQMessageListener(consumerGroup = "dk2202",topic = "tx-topic")
public class RocketMQConsumer implements RocketMQListener, RocketMQPushConsumerLifecycleListener {

    @Autowired
    StringRedisTemplate stringRedisTemplate;


    @DubboReference
    UserScoreService us;


    @Override
    public void onMessage(Object o) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //接收我们mq的消息
                MessageExt messageExt=list.get(0);
                int reconsumeTimes = messageExt.getReconsumeTimes();//重试次数
                System.out.println("重试次数为"+reconsumeTimes);
                try {
                    messageExt.getTopic();//接收topic
                    messageExt.getMsgId();//消息id
                    //判断传来的消息是否存在，redis中建一个id表
                    String msgId = messageExt.getMsgId();
                    Boolean hasKey = stringRedisTemplate.hasKey("MESSAGE" + msgId);
                    if(!hasKey){
                        byte[] body = messageExt.getBody();//消息内容
                        String msg= new String (body);
                        System.out.println("消息内容为"+msg);
                        userscoreVO userscoreVO = JSONArray.parseObject(msg, userscoreVO.class);//转对象
//                        System.out.println(userscoreVO.money);
//                        System.out.println(userscoreVO.userId);
                        TbUserScore ts=new TbUserScore();
                        ts.setUserid(userscoreVO.userId);//用户名
                        String strDouble = String.valueOf(userscoreVO.money);//double转string
                        ts.setIntegra(strDouble);//现有积分
                        ts.setCreateTime(new Date());
                        us.addScore(ts);


//                        int x=9/0;
                        stringRedisTemplate.boundValueOps("MESSAGE"+msgId).set("已消费");
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//消费成功返回

                }catch (Exception e){
                    e.printStackTrace();
                    if(reconsumeTimes<2){//如果重试次数小于2继续重试，大于2就消费成功，放入死信队列
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//消费失败重试
                    }
                    System.out.println("消息重试次数>2次，加入死信队列，等待人工补偿");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//消费成功返回
                }

            }
        });
    }
}




